package com.gitee.melin.bee.core.jdbc.relational;

import com.gitee.melin.bee.core.jdbc.enums.CompatibleMode;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.enums.SaslMchanism;
import com.gitee.melin.bee.util.JsonUtils;
import java.util.Map;
import lombok.Builder;
import lombok.Getter;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.commons.lang3.builder.HashCodeBuilder;
import org.apache.hadoop.conf.Configuration;
import org.codehaus.jackson.annotate.JsonIgnore;

@Getter
public class ConnectionInfo {

    // hive 数据源配置
    private static final Configuration DEFAULT_CONFIGURATION = new Configuration();

    static {
        DEFAULT_CONFIGURATION.set("hadoop.security.authentication", "kerberos");
        DEFAULT_CONFIGURATION.set("hadoop.security.authorization", "true");
    }

    // NONE, PASSWORD, KERBEROS, SASL_PLAINTEXT, SASL_SSL, SSL
    private String protocol;

    // kafka 认真场景：PLAIN, SCRAM-SHA-256, SCRAM-SHA-512，GSSAPI
    private SaslMchanism saslMchanism;

    // kafka sasl jass config
    private String jaasConfig;

    private String code;

    private DataSourceType dataSourceType;

    private String username;

    @JsonIgnore
    private String password;

    private String url;

    private String accessKeyId;

    @JsonIgnore
    private String secretAccessKey;

    private String endpoint;

    private String databaseName;

    private String hivePrincipal;

    private String principal;

    private String keytabFile;

    private String krb5File;

    private Integer minimumIdle = 1;

    private Integer maxPoolSize = 20;

    private Map<String, String> properties;

    private Integer queryTimeout; // seconds

    // oceabbase 数据库兼容模式
    private CompatibleMode compatibleMode;

    // keytabFile 文件md5
    private String keytabFileMd5;

    // krb5File 文件md5
    private String krb5FileMd5;

    private Configuration configuration;

    // kafka ssl 配置参数
    private String sslKeyPassword;

    private String sslTruststoreLocation;

    private String sslTruststorePassword;

    private String sslKeystoreLocation;

    private String sslKeystorePassword;

    // sasl mechanism = gssapi
    private String servicePrincipal;

    public ConnectionInfo() {}

    @Builder(setterPrefix = "with")
    public ConnectionInfo(
            String protocol,
            SaslMchanism saslMchanism,
            String code,
            DataSourceType dataSourceType,
            String username,
            String password,
            String url,
            String accessKeyId,
            String secretAccessKey,
            String endpoint,
            String databaseName,
            String hivePrincipal,
            String principal,
            String keytabFile,
            String krb5File,
            Integer minimumIdle,
            Integer maxPoolSize,
            Map<String, String> properties,
            Integer queryTimeout,
            CompatibleMode compatibleMode,
            Configuration configuration,
            String sslKeyPassword,
            String sslTruststoreLocation,
            String sslTruststorePassword,
            String sslKeystoreLocation,
            String sslKeystorePassword,
            String jaasConfig,
            String servicePrincipal) {

        if (StringUtils.isBlank(protocol)) {
            this.protocol = "None";
        } else {
            this.protocol = protocol;
        }
        this.saslMchanism = saslMchanism;
        this.code = StringUtils.trim(code);
        this.dataSourceType = dataSourceType;
        this.username = StringUtils.trim(username);
        this.password = password;
        this.url = StringUtils.trim(url);
        this.accessKeyId = accessKeyId;
        this.secretAccessKey = secretAccessKey;
        this.endpoint = StringUtils.trim(endpoint);
        this.databaseName = databaseName;
        this.hivePrincipal = StringUtils.trim(hivePrincipal);
        this.principal = StringUtils.trim(principal);
        this.keytabFile = keytabFile;
        this.krb5File = krb5File;
        if (minimumIdle != null) {
            this.minimumIdle = minimumIdle;
        }
        if (maxPoolSize != null) {
            this.maxPoolSize = maxPoolSize;
        }
        this.properties = properties;
        this.queryTimeout = queryTimeout;
        this.compatibleMode = compatibleMode;
        this.configuration = configuration;

        // kafka ssl
        this.sslKeyPassword = sslKeyPassword;
        this.sslTruststoreLocation = sslTruststoreLocation;
        this.sslTruststorePassword = sslTruststorePassword;
        this.sslKeystoreLocation = sslKeystoreLocation;
        this.sslKeystorePassword = sslKeystorePassword;
        this.jaasConfig = jaasConfig;
        this.servicePrincipal = servicePrincipal;

        /*if (StringUtils.isNotBlank(krb5File)) {
            try (InputStream is = Files.newInputStream(Paths.get(krb5File))) {
                krb5FileMd5 = DigestUtils.md5Hex(is);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

        if (StringUtils.isNotBlank(keytabFile)) {
            try (InputStream is = Files.newInputStream(Paths.get(keytabFile))) {
                keytabFileMd5 = DigestUtils.md5Hex(is);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }*/
    }

    public Configuration getConfiguration() {
        if (configuration == null) {
            return DEFAULT_CONFIGURATION;
        } else {
            return configuration;
        }
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }

        if (o == null || getClass() != o.getClass()) {
            return false;
        }

        ConnectionInfo that = (ConnectionInfo) o;

        return new EqualsBuilder()
                .append(protocol, that.protocol)
                .append(saslMchanism, that.saslMchanism)
                .append(dataSourceType, that.dataSourceType)
                .append(username, that.username)
                .append(password, that.password)
                .append(url, that.url)
                .append(accessKeyId, that.accessKeyId)
                .append(secretAccessKey, that.secretAccessKey)
                .append(endpoint, that.endpoint)
                .append(databaseName, that.databaseName)
                .append(hivePrincipal, that.hivePrincipal)
                .append(principal, that.principal)
                .append(jaasConfig, that.jaasConfig)
                .append(servicePrincipal, that.servicePrincipal)
                .append(minimumIdle, that.minimumIdle)
                .append(maxPoolSize, that.maxPoolSize)
                .append(properties, that.properties)
                .append(queryTimeout, that.queryTimeout)
                .append(compatibleMode, that.compatibleMode)
                .isEquals();
    }

    @Override
    public int hashCode() {
        return new HashCodeBuilder(17, 37)
                .append(protocol)
                .append(saslMchanism)
                .append(dataSourceType)
                .append(username)
                .append(password)
                .append(url)
                .append(accessKeyId)
                .append(secretAccessKey)
                .append(endpoint)
                .append(databaseName)
                .append(hivePrincipal)
                .append(principal)
                .append(jaasConfig)
                .append(servicePrincipal)
                .append(minimumIdle)
                .append(maxPoolSize)
                .append(properties)
                .append(queryTimeout)
                .append(compatibleMode)
                .toHashCode();
    }

    @Override
    public String toString() {
        return JsonUtils.toJSONString(this);
    }
}
